package cool.taomu.toolkit.kademlia.service

import com.google.common.base.Function
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.ListeningExecutorService
import com.google.common.util.concurrent.MoreExecutors
import com.google.common.util.concurrent.RateLimiter
import com.google.inject.Inject
import cool.taomu.toolkit.impl.StoreKeyValue
import cool.taomu.toolkit.inter.IFunction
import cool.taomu.toolkit.inter.IStoreKeyValue
import cool.taomu.toolkit.jersey.JerseyInternalInvoker
import cool.taomu.toolkit.jersey.RequestDto
import cool.taomu.toolkit.kademlia.entity.KademliaNode
import cool.taomu.toolkit.kademlia.entity.KademliaReponse
import cool.taomu.toolkit.kademlia.entity.ServiceNode
import cool.taomu.toolkit.kademlia.utils.KademliaUtils
import cool.taomu.toolkit.kademlia.utils.ResourceCollector
import java.util.ArrayList
import java.util.Arrays
import java.util.List
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.stream.Collectors
import javax.inject.Named
import org.apache.thrift.TException
import org.eclipse.xtend.lib.annotations.Accessors
import org.slf4j.LoggerFactory

@Accessors
class KademliaSerivce {
	val static LOG = LoggerFactory.getLogger(KademliaSerivce);

	@Inject
	RoutingTable table;

	@Inject(optional=true)
	@Named("taomu.kademlia.k")
	int k = 8;

	@Inject
	ResourceCollector collector;

	@Inject
	JerseyInternalInvoker invoker;

	ExecutorService executor = Executors.newCachedThreadPool();
	ListeningExecutorService guavaExecutor = MoreExecutors.listeningDecorator(executor);

	@Inject(optional=true)
	IStoreKeyValue<String, byte[]> store = new StoreKeyValue<String, byte[]>();

	new() {
	}

	@Accessors
	static class findNodeEntity {
		/**
		 * 查找的nodeId
		 */
		String nodeId;
		/**
		 * 传入的节点
		 */
		KademliaNode node
		/**
		 * k桶中的节点(要查询的节点)
		 * 
		 */
		KademliaNode knode

		new(String nodeId, KademliaNode node, KademliaNode knode) {
			this.nodeId = nodeId;
			this.node = node;
			this.knode = knode;
		}
	}

	/**
	 * @Param nodeId 查找的节点ID
	 * @Param node 节点内容
	 * @Param findNodeFunc 通过网络查询节点
	 * @Param checkNode 验证节点是否联通
	 */
	def synchronized findNode(String nodeId, KademliaNode node,
		Function<findNodeEntity, List<KademliaNode>> findNodeFunc,
		Function<KademliaNode, Boolean> checkNode) throws TException {
		LOG.info("****执行findNode**** node:{} \n==>\n localNode:{}", node, table.localNode)
		LOG.info("****buckets大小:{}****", table.bucketsTotalSize)
		if (node !== null && !node.equals(table.localNode)) {
			guavaExecutor.submit([
				table.addNode(node, checkNode);
			])
		}
		val nodes = table.findClosestNodes(nodeId, this.k);
		LOG.info("查找到的节点数量:{}", nodes.size);
		val ListenableFutures = new ArrayList<ListenableFuture<ArrayList<KademliaNode>>>();
		nodes.forEach [ knode |
			if (!knode.equals(node)) {
				var futures = guavaExecutor.submit([
					val resNodes = new ArrayList<KademliaNode>();
					try {
						LOG.info("连接远程服务器查找节点:{}", knode);
						resNodes.addAll(findNodeFunc.apply(new findNodeEntity(nodeId, node, knode)))
					} catch (Exception ex) {
						LOG.error("连接服务器错误:{}", ex);
					}
					LOG.info("查找到节点的大小:{}", resNodes.size);
					return resNodes;
				])
				ListenableFutures.add(futures);
			}
		]
		LOG.info("执行查找节点的线程数:{}", ListenableFutures.size);
		var listListenableFuture = Futures.allAsList(ListenableFutures);
		// 将每个 ArrayList<Node> 转换为流并合并
		var closestNodes = listListenableFuture.get().stream().flatMap([it.stream()]).distinct().collect(
			Collectors.toList());
		LOG.info("返回节点数:{}", closestNodes.size);
		if (closestNodes.size == 0) {
			return closestNodes
		}
		// 将找到的节点加入到自己的桶中
		LOG.info("需要将找到的{}个节点加入到自己的桶中", closestNodes.size);
		closestNodes.forEach [
			guavaExecutor.submit([
				table.addNode(it, checkNode);
			])
		]
		closestNodes.sort([ n1, n2 |
			RoutingTable.xorDistance(nodeId, n1.getId()).compareTo(RoutingTable.xorDistance(nodeId, n2.getId()))
		])
		var resNodes = closestNodes.subList(0, Math.min(this.k, closestNodes.size()));
		LOG.info("最终返回节点数:{}", resNodes.size);
		return resNodes
	}

	private def findRemoteNode(String key, List<KademliaNode> nodes, List<KademliaNode> node,
		Function<KademliaNode, List<KademliaReponse>> findRemoteValue) {
		val resList = new ArrayList<KademliaReponse>();
		LOG.info("远程节点数:{}", nodes.size)
		nodes.forEach [ knode |
			if (node.filterNull.findFirst[it.id == knode.id] === null) {
				resList.addAll(findRemoteValue.apply(knode));
				if (resList.size > 0) {
					LOG.info("返回找到key:{},对应的值,size:{}", key, resList.size);
					return;
				}
			}
		]
		return resList;
	}

	def synchronized findValue(String key, KademliaNode node,
		Function<KademliaNode, List<KademliaReponse>> findRemoteValue,
		Function<KademliaNode, Boolean> checkNode) throws TException {
		LOG.info("****执行findValue****")
		LOG.info("****buckets大小:{}****", table.bucketsTotalSize)
		if (node !== null && !node.equals(table.localNode)) {
			guavaExecutor.submit([
				table.addNode(node, checkNode);
			])
		}
		var key160 = KademliaUtils.stringTo160Bit(key);
		LOG.info("****key160****:{}", key160)
		if (store.containsKey(key160)) {
			LOG.info("****返回findValue存储的数据****")
			if (node === null) {
				return new KademliaReponse(null, store.get(key160), 200);
			} else {
				return new KademliaReponse(Arrays.asList(node), store.get(key160), 200);
			}
		}
		LOG.info("从K桶中获取节点")
		val nodes = table.findClosestNodes(key160, this.k);
		LOG.info("节点数:{}", nodes.size)
		if (!nodes.empty) {
			val resList = new ArrayList<KademliaReponse>();
			resList.addAll(this.findRemoteNode(key, nodes, Arrays.asList(node, table.localNode), findRemoteValue));
			if (resList.size > 0) {
				return resList.get(0);
			}
		}
		LOG.info("返回未找到节点")
		return new KademliaReponse(null, "404", 404);
	}

	val rateLimiter = RateLimiter.create(2.0)

	private def findRemoteServiceNode(String key, List<KademliaNode> nodes, List<KademliaNode> node,
		Function<KademliaNode, List<KademliaReponse>> findRemoteService) {
		val resList = new ArrayList<KademliaReponse>();
		LOG.info("远程节点数:{}", nodes.size)
		// 远程节点每次只执行一个
		nodes.forEach [ knode |
			if (node.filterNull.findFirst[it.id == knode.id] === null) {
				resList.addAll(findRemoteService.apply(knode));
				if (resList.size > 0) {
					LOG.info("返回找到key:{},对应的值,size:{}", key, resList.size);
					return;
				}
			}
		]
		return resList;
	}

	/**
	 * @Param key 服务名
	 * @Param value 服务调用参数
	 * @Param node 服务节点
	 * @Param findRemoteService 查找远程服务
	 * @Param checkNode 检查节点是否可用
	 */
	def synchronized findService(RequestDto dto, KademliaNode node,
		Function<KademliaNode, List<KademliaReponse>> findRemoteService,
		Function<KademliaNode, Boolean> checkNode) throws TException {
		LOG.info("****执行findService****")
		LOG.info("****buckets大小:{}****", table.bucketsTotalSize)
		// 500毫秒内，没拿到令牌，就直接进入服务降级
		if (node !== null && !node.equals(table.localNode)) {
			guavaExecutor.submit([
				table.addNode(node, checkNode);
			])
		}
		var key = #[dto.method, dto.serviceName].join(":");
		LOG.info("***Key****:{}", key)
		var boolean tryAcquire = rateLimiter.tryAcquire(500, TimeUnit.MILLISECONDS);
		var key160 = KademliaUtils.stringTo160Bit(key);
		LOG.info("****key160****:{}", key160)
		if (collector.scanResult.contains(#[dto.method, dto.serviceName].join(":")) && tryAcquire) {
			LOG.info("****返回findService执行的结果1****")
			var ins = invoker.invoke(dto);
			if (node === null) {
				return new KademliaReponse(null, ins, 200);
			} else {
				return new KademliaReponse(Arrays.asList(node), ins, 200);
			}
		}
		LOG.info("从K桶中获取节点")
		val nodes = table.findClosestNodes(key160, this.k);
		LOG.info("节点数:{}", nodes.size)
		if (!nodes.empty) {
			val resList = new ArrayList<KademliaReponse>();
			resList.addAll(
				this.findRemoteServiceNode(key, nodes, Arrays.asList(node, table.localNode), findRemoteService));
			if (resList.size > 0) {
				return resList.get(0);
			}
		}
		LOG.info("在网络上未找到访问节点，再次判断本地是否有服务：{}", key)
		if (!tryAcquire) {
			rateLimiter.acquire();
		}
		if (collector.scanResult.contains(key) && tryAcquire) {
			LOG.info("****返回findService执行的结果2****")
			var ins = invoker.invoke(dto);
			if (node === null) {
				return new KademliaReponse(null, ins, 200);
			} else {
				return new KademliaReponse(Arrays.asList(node), ins, 200);
			}
		}
		LOG.info("未找到服务：{}", key);
		return new KademliaReponse(null, "404", 404);
	}

	def synchronized ping(KademliaNode knode, Function<KademliaNode, Boolean> checkNode) throws TException {
		LOG.info("****执行ping****:{}", knode)
		LOG.info("****buckets大小:{}****", table.bucketsTotalSize)
		guavaExecutor.submit([
			table.addNode(knode, checkNode);
		])
		return new KademliaReponse(null, "PONG", 200);
	}

	def synchronized store(
		String key,
		String value,
		Function<findNodeEntity, List<KademliaNode>> findNodeFunc,
		IFunction<ServiceNode> remoteStoreFunc,
		Function<KademliaNode, Boolean> remoteNode
	) throws TException {
		LOG.info("****执行store****")
		LOG.info("****buckets大小:{}****", table.bucketsTotalSize)
		LOG.info("存储key:{}的值", key);
		var key160 = KademliaUtils.stringTo160Bit(key);
		if (key160.equals(table.localNode.id)) {
			store.put(key160, value.getBytes("UTF-8"));
		} else {
			var nodes = this.findNode(key160, table.localNode, findNodeFunc, remoteNode);
			if (nodes.size > 0) {
				LOG.info("找到的节点数:{}", nodes.size);
				val ListenableFutures = new ArrayList<ListenableFuture<?>>();
				nodes.forEach [ knode |
					ListenableFutures.add(guavaExecutor.submit([
						try {
							remoteStoreFunc.apply(new ServiceNode(knode, key, value));
						} catch (Exception ex) {
							ex.printStackTrace
							return 0;
						}
						return 1;
					]))
				]
				LOG.info("执行store操作的线程数:{}", ListenableFutures.size);
				var listListenableFuture = Futures.allAsList(ListenableFutures);
				var result = listListenableFuture.get();
				LOG.info("执行store操作结束:{}", result);
			} else {
				LOG.info("未找到相邻节点,将数据存下来");
				store.put(key160, value.getBytes("UTF-8"));
			}
		}
	}
}
